package org.company.mars.kafka;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import org.junit.Test;

public class ConsumerDemo {
	
	@Test
	public void simpleConsume() {
		Properties props = new Properties();
		props.put("zookeeper.connect", "10.86.44.167:2181");
        props.put("group.id", "test");
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
		
		ConsumerConfig config = new ConsumerConfig(props);
		ConsumerConnector consumer = 
				Consumer.createJavaConsumerConnector(config);
		
		String topic = "test1";
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, 1);
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap
			= consumer.createMessageStreams(topicCountMap);
		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
		for(final KafkaStream<byte[],byte[]> stream : streams) {
			ConsumerIterator<byte[], byte[]> it = stream.iterator();
			while(it.hasNext()) {
				System.out.println(new String(it.next().message()));
			}
		}
	}
	
}